package com.heima.article.listener;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.heima.article.service.IApArticleConfigService;
import com.heima.article.service.IApArticleContentService;
import com.heima.article.service.IApArticleService;
import com.heima.common.constants.Constant;
import com.heima.common.constants.KafkaConstant;
import com.heima.model.acticle.pojos.ApArticleConfig;
import com.heima.model.acticle.pojos.ApArticleContent;
import freemarker.template.TemplateException;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.Map;
import java.util.stream.Collectors;

@Component
@Slf4j
public class ArticleListener {

    @Resource
    private IApArticleService iApArticleService;
    @Resource
    private IApArticleConfigService iApArticleConfigService;
    @Resource
    private IApArticleContentService iApArticleContentService;

    @KafkaListener(topics = KafkaConstant.TOPIC_ENABLE_DOWN)
    @Transactional
    public void onEnable(ConsumerRecord<String, String> record, Consumer<String,String> consumer){
        if (null == record){
            log.info("没有收到消息");
            return;
        }
        //
        Map<String,String> parse = JSON.parseObject(record.value(), Map.class);
        Long id = Long.parseLong(parse.get(KafkaConstant.MSG_ENABLE_DOWN_ID));
        Integer enable = Integer.parseInt(parse.get(KafkaConstant.MSG_ENABLE_DOWN_ENABLE));
        //
        consumer.commitSync();
        //
        if (enable.equals(Constant.NUMBER_ONE)){
            LambdaUpdateWrapper<ApArticleConfig> wrapper = Wrappers.<ApArticleConfig>lambdaUpdate()
                    .set(ApArticleConfig::getIsDown,Constant.NUMBER_ZERO)
                    .eq(ApArticleConfig::getArticleId, id);
            iApArticleConfigService.update(wrapper);
        }else {
            LambdaUpdateWrapper<ApArticleConfig> wrapper = Wrappers.<ApArticleConfig>lambdaUpdate()
                    .set(ApArticleConfig::getIsDown,Constant.NUMBER_ONE)
                    .eq(ApArticleConfig::getArticleId, id);
            iApArticleConfigService.update(wrapper);
        }

    }

    @KafkaListener(topics = KafkaConstant.TOPIC_ARTICLE_UP)
    @Transactional
    public void onUpArticle(ConsumerRecord<String, String> record, Consumer<String,String> consumer) throws IOException, TemplateException {
        if (null == record){
            log.info("没有收到消息");
            return;
        }
        //
        long id = Long.parseLong(record.value());
        //
        consumer.commitSync();
        //
        iApArticleService.upArticle(id);
    }
}
